package com.lg.controller;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import com.lg.service.KafkaService;
import com.system.constants.QueueConstant;

import java.util.concurrent.ExecutionException;

/**
 * 消费者消费失败处理-重试队列
 * kafka没有重试机制不支持消息重试，也没有死信队列，因此使用kafka做消息队列时，需要自己实 现消息重试的功能。
 * 实现：创建新的kafka主题作为重试队列：
 * 1.创建一个topic作为重试topic，用于接收等待重试的消息。
 * 2.普通topic消费者设置待重试消息的下一个重试topic。
 * 3.从重试topic获取待重试消息储存到redis的zset中，并以下一次消费时间排序
 * 4.定时任务从redis获取到达消费事件的消息，并把消息发送到对应的topic
 * 5.同一个消息重试次数过多则不再重试
 * 
 * @author kaifa002
 *
 */
@RestController
public class RetryController {

	@Autowired
	private KafkaService kafkaService;

	@RequestMapping("/send/{message}")
	public String sendMessage(@PathVariable String message) throws ExecutionException, InterruptedException {

		ProducerRecord<String, String> record = new ProducerRecord<>(QueueConstant.PROC_TEST_TOPIC, message);

		// 向业务主题发送消息
		String result = kafkaService.sendMessage(record);

		return result;
	}

}